package vip.huhailong.redismq.redistool;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.stream.StreamListener;
import vip.huhailong.redismq.service.ISegmentService;
import vip.huhailong.redismq.service.impl.ServiceFactory;

/**
 * @author Huhailong
 * @Description 监听消息处理类
 * @Date 2021/3/10.
 */
@Slf4j
public class ListenerMessage implements StreamListener<String, MapRecord<String, String, String>> {

    RedisUtil redisUtil;

    public ListenerMessage(RedisUtil redisUtil){
        this.redisUtil = redisUtil;
    }

    @Override
    public void onMessage(MapRecord<String, String, String> entries) {

        try{
            System.out.println("----------------------------------------------------------");
            System.out.println(entries);
            System.out.println(entries.getStream());
            System.out.println(entries.getValue().get("test"));
            //check用于验证key和对应消息是否一直
            log.info("stream name :{}, body:{}, check:{}",entries.getStream(),entries.getValue(),(entries.getStream().equals(entries.getValue().get("test"))));
            String engineType="offline";
            //处理业务逻辑
            ISegmentService service = ServiceFactory.getService(engineType);
            //添加分布锁，防止多节点重复执行 todo 未完成
            String taskId = entries.getStream();
            String segmentId = entries.getStream();
            String jsonStr = entries.getStream();
            service.segmentHandle(taskId,segmentId,jsonStr);

            redisUtil.delField(entries.getStream(),entries.getId().getValue());
            System.out.println("----------------------------------------------------------");
        }catch (Exception e){
            log.error("error message:{}",e.getMessage());
        }finally {
            // 消费完成后确认消费（ACK） todo 未完成
//            this.ackMessage(stream,recordVal);
            System.out.println("消费完成后确认消费（ACK）");
        }
    }

    // 添加分布锁，防止多节点重复执行 todo 未完成
//    private void unlock(String lockKey,String lockVal){
//        //释放锁
//        try{
//            redisUtil.unlock(lockKey,lockVal,0, TimeUnit.MINUTES);
//            log.info("释放分布式锁成功！lockKey={}",lockKey);
//        }catch (Exception ex){
//            log.error("释放分布式锁异常！lockKey={}",lockKey);
//        }
//    }
    // 消费完成后确认消费（ACK） todo 未完成
//    private void ackMessage(String stream,String recordVal){
//        Long num = redisUtil.ack(stream,group, recordVal);
//        if(num != null && num > 0){
//            log.info("消息确认成功！stream={}, group={}, recordId={}",stream,group,recordVal);
//        }
//        //删除已确认消息
//        redisUtil.delField(stream,recordVal);
//        log.info("消息删除成功！stream={}, group={}, recordId={}",stream,group,recordVal);
//    }


}
